package com.shujia.spark.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo24Broadcast {
  def main(args: Array[String]): Unit = {
    // 广播变量：主要应用在MapJoin中，适合大表关联小表，将小表广播

    // 关联学生和分数的数据

    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName(this.getClass.getSimpleName.replace("$", ""))
    val sc = new SparkContext(conf)

    val stuRDD: RDD[String] = sc.textFile("spark/data/stu/students.txt")
    val scoreRDD: RDD[String] = sc.textFile("spark/data/stu/score.txt", 8)

    val stuKVRDD: RDD[(String, String)] = stuRDD.map(line => (line.split(",")(0), line))
    val scoreKVRDD: RDD[(String, String)] = scoreRDD.map(line => (line.split(",")(0), line))

    // 第一种方式：使用join类算子
    scoreKVRDD.join(stuKVRDD).map(kv => {
      val id: String = kv._1
      val scoreStr: String = kv._2._1
      val stuStr: String = kv._2._2
      s"$id,$scoreStr,$stuStr"
    }).foreach(println)

    // 第二种方式：使用集合Map进行关联
    // 类似MR中的MapJoin：不需要reduce任务就可以完成join，避免shuffle提高整体性能
    val stuMap: collection.Map[String, String] = stuKVRDD.collectAsMap()
    // 使用广播变量，将Driver端定义的变量进行广播，广播给每个Executor
    val stuMapBro: Broadcast[collection.Map[String, String]] = sc.broadcast(stuMap)

    scoreKVRDD
      .map(kv => {
        val id: String = kv._1
        val scoreStr: String = kv._2
        // 基于stuMap通过id获取value实现关联
        // RDD分区数为多少，Task的数量就有多少，同时stuMap就会被复制N次，最终Task会被发送到Executor中执行
        // Task的数量 >> Executor的数量，可以考虑给每个Executor发送一个stuMap，减少复制的次数，提升效率
        //        val stuStr: String = stuMap.getOrElse(id, "")
        // 通过广播的变量.value可以直接从Executor获取到广播的变量
        val sMap: collection.Map[String, String] = stuMapBro.value
        val stuStr: String = sMap.getOrElse(id, "")
        // RDD内部是不能直接嵌套另一个RDD的
        //        val stuStr2: String = stuKVRDD.collectAsMap().getOrElse(id, "")
        s"$id,$scoreStr,$stuStr"
      }).foreach(println)

    while (true) {

    }
  }

}
